package com.wechat.network;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.exception.RemotingException;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.Message;
import com.swak.router.HandlerFunction;
import com.swak.router.RouterFunction;
import com.wechat.command.Command;
import com.wechat.command.CommandResult;
import com.wechat.service.cmds.CmdRoute;

/**
 * 网络 - 处理器
 * 
 * @author 618lf
 */
public interface NetworkHandler {

	/**
	 * 设置处理器
	 * 
	 * @param router
	 */
	void bind(RouterFunction<CmdRoute> router);

	/**
	 * on channel connected.
	 *
	 * @param channel channel.
	 */
	default void connected(NetworkChannel channel) throws RemotingException {
	}

	/**
	 * on channel disconnected.
	 *
	 * @param channel channel.
	 */
	default void disconnected(NetworkChannel channel) throws RemotingException {
	}

	/**
	 * on message sent.
	 *
	 * @param channel channel.
	 * @param message message.
	 */
	default void sent(NetworkChannel channel, Object message) throws RemotingException {
	}

	/**
	 * on exception caught.
	 *
	 * @param channel   channel.
	 * @param exception exception.
	 */
	default void caught(NetworkChannel channel, Throwable exception) throws RemotingException {
	}

	/**
	 * reply.
	 *
	 * @param channel
	 * @param message
	 * @return response
	 * @throws RemotingException
	 */
	CompletionStage<Object> reply(NetworkChannel channel, Object msg) throws RemotingException;

	/**
	 * on message received.
	 *
	 * @param channel channel.
	 * @param message message.
	 */
	void received(NetworkChannel channel, Object msg) throws RemotingException;

	/**
	 * create NetworkHandler.
	 * 
	 * @return
	 */
	NetworkHandler handler = new RouterNetworkHandler();

	public static NetworkHandler newNetworkHandler() {
		return handler;
	}

	/**
	 * Command => Handler
	 * 
	 * @author 618lf
	 */
	static class RouterNetworkHandler implements NetworkHandler {

		private volatile RouterFunction<CmdRoute> router;
		private final HandlerFunction<CmdRoute> DefHandler = (res) -> CommandResult.SUCCESS;

		private HandlerFunction<CmdRoute> route(CmdRoute route) {
			if (this.router != null) {
				return this.router.route(route);
			}
			return DefHandler;
		}

		@Override
		public void bind(RouterFunction<CmdRoute> router) {
			this.router = router;
		}

		@SuppressWarnings("unchecked")
		private CompletionStage<Object> invoke(NetworkChannel channel, Command command) {
			CmdRoute route = new CmdRoute(channel, command);
			HandlerFunction<CmdRoute> handler = this.route(route);

			Object result = handler.handle(route);
			if (result != null && result instanceof CompletionStage) {
				return ((CompletableFuture<Object>) result);
			}
			return CompletableFuture.completedFuture(result);
		}

		@Override
		public CompletionStage<Object> reply(NetworkChannel channel, Object msg) throws RemotingException {
			Object data = ((Message) msg).getData();
			if (data == null || !(data instanceof Command)) {
				return CompletableFuture.completedFuture(null);
			}
			return this.invoke(channel, (Command) data);
		}

		@Override
		public void received(NetworkChannel channel, Object msg) throws RemotingException {
			Object data = ((Message) msg).getData();
			if (data == null || !(data instanceof Command)) {
				return;
			}

			this.invoke(channel, (Command) data);
		}
	}

	/**
	 * ExchangeHandler => NetworkHandler
	 * 
	 * @author 618lf
	 */
	public static class NetworkExchangeHandler implements ExchangeHandler {

		Logger logger = LoggerFactory.getLogger(Network.class);

		protected final Network network;
		protected final NetworkHandler delegate;

		public NetworkExchangeHandler(Network network, NetworkHandler delegate) {
			this.network = network;
			this.delegate = delegate;
		}

		@Override
		public void connected(Channel channel) throws RemotingException {
			NetworkChannel exchangeChannel = NetworkChannel.getOrAddChannel(network, channel);
			this.delegate.connected(exchangeChannel);
		}

		@Override
		public void disconnected(Channel channel) throws RemotingException {
			NetworkChannel exchangeChannel = NetworkChannel.getOrAddChannel(network, channel);
			try {
				delegate.disconnected(exchangeChannel);
			} finally {
				NetworkChannel.removeChannel(channel);
			}
		}

		@Override
		public void sent(Channel channel, Object message) throws RemotingException {
			try {
				NetworkChannel exchangeChannel = NetworkChannel.getOrAddChannel(network, channel);
				delegate.sent(exchangeChannel, message);
			} catch (Throwable t) {
				NetworkChannel.removeChannelIfDisconnected(channel);
			}
		}

		@Override
		public void received(Channel channel, Object message) throws RemotingException {
			if (logger.isDebugEnabled()) {
				logger.debug("[网络] 接收命令 {}", message);
			}
			NetworkChannel exchangeChannel = NetworkChannel.getOrAddChannel(network, channel);
			delegate.received(exchangeChannel, message);
		}

		@Override
		public void caught(Channel channel, Throwable exception) throws RemotingException {
			NetworkChannel exchangeChannel = NetworkChannel.getOrAddChannel(network, channel);
			try {
				delegate.caught(exchangeChannel, exception);
			} finally {
				NetworkChannel.removeChannelIfDisconnected(channel);
			}
		}

		@Override
		public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
			if (logger.isDebugEnabled()) {
				logger.debug("[网络] 接收命令 {}", message);
			}

			NetworkChannel exchangeChannel = NetworkChannel.getOrAddChannel(network, channel);
			return this.delegate.reply(exchangeChannel, message);
		}
	}
}
